package confirm;

import com.rabbitmq.client.*;
import config.RabbitConfig;

import java.io.IOException;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

/**
 * 异步确认
 *
 * @author tuyou
 * @date 2022/6/14 22:38
 */
public class AsynConfirm {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        try (Connection connection = RabbitConfig.getConnection()) {

            Channel channel = connection.createChannel();
            String queue = channel.queueDeclare().getQueue();
            int count = 10;

            //本地缓存集合，用于失败重试
            TreeSet unConfirmSet = new TreeSet();
            channel.confirmSelect();
            //添加服务器确认监听器，这样就不用手动调用channel.waitForConfirms()阻塞等待
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    System.out .println( "ack, SeqNo : " + deliveryTag + ", multiple : " + multiple) ;
                    System.out.println("before set:" + unConfirmSet);
                    if (multiple) {
                        unConfirmSet.headSet(deliveryTag + 1).clear();
                    } else {
                        unConfirmSet.remove(deliveryTag);
                    }
                    System.out.println("after set:" + unConfirmSet);
                }

                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    System.out .println( "nack, SeqNo : " + deliveryTag + ", multiple : " + multiple) ;
                    System.out.println("before set:" + unConfirmSet);
                    if (multiple) {
                        unConfirmSet.headSet(deliveryTag + 1).clear();
                    } else {
                        unConfirmSet.remove(deliveryTag);
                    }
                    System.out.println("after set:" + unConfirmSet);
                    //处理消息重发
                    System.out.println("消息重发");
                }
            });
            for (int i = 0; i < count; i++) {
                long seqNo = channel.getNextPublishSeqNo();
                channel.basicPublish("", queue, MessageProperties.PERSISTENT_TEXT_PLAIN, String.valueOf(i).getBytes());
                unConfirmSet.add(seqNo);
            }

            for (int i = 0; i < count; i++) {
                GetResponse resp = channel.basicGet(queue, true);
                System.out.println(new String(resp.getBody()));
            }
        }
    }
}
